package net.bwie.gd4.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class KafkaJob3 {
    public static void main(String[] args) {
        //1-表的执行环境
        TableEnvironment tabEnv = getTableEnv();

        //2-输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

////
        //3-数据处理-select
        handle(tabEnv);
////
        //4-输出表-output：映射到doris表
        createOutputTable(tabEnv);
////
        //5-将数据保存到doris
        saveToDoris(tabEnv) ;
    }


    private static void saveToDoris(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "INSERT INTO dwd_user_product_log_doris_sink\n" +
                        "select\n" +
                        "    user_id ,\n" +
                        "    behavior_id  ,\n" +
                        "    behavior_type,\n" +
                        "    category_id,\n" +
                        "    item_id,\n" +
                        "    behavior_time,\n" +
                        "    age_related,\n" +
                        "    tags\n" +
                        "from dwd_user_product_log"
        );



        tabEnv.executeSql(
                "INSERT INTO dim_pet_categorys_doris_sink\n" +
                        "select\n" +
                        "    category_id ,\n" +
                        "    category_name ,\n" +
                        "    pet_type,\n" +
                        "    is_common,\n" +
                        "    age_stage\n" +
                        "from dim_pet_categorys"
        );

    }






    private static void createOutputTable(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "CREATE TABLE dwd_user_product_log_doris_sink(\n" +
                        "    `user_id` bigint  COMMENT '用户ID',\n" +
                        "    `behavior_id` STRING ,\n" +
                        "    `behavior_type` STRING,\n" +
                        "    `category_id` bigint,\n" +
                        "    `item_id` string,\n" +
                        "    `behavior_time` timestamp,\n" +
                        "    `age_related` bigint,\n" +
                        "    `tags` STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'gongdan13.dwd_user_product_log',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.batch.interval' = '10s',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.batch.size' = '1000'\n" +
                        ")"
        );


        tabEnv.executeSql(
                "CREATE TABLE dim_pet_categorys_doris_sink(\n" +
                        "    `category_id` bigint ,\n" +
                        "    `category_name` STRING ,\n" +
                        "    `pet_type` STRING,\n" +
                        "    `is_common` bigint,\n" +
                        "    `age_stage` string\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'gongdan13.dim_pet_categorys',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.batch.interval' = '10s',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.batch.size' = '1000'\n" +
                        ")"
        );



    }


    private static void handle(TableEnvironment tabEnv) {


        tabEnv.executeSql(
                "CREATE TABLE `dim_pet_categorys`(\n" +
                        "    `category_id` bigint ,\n" +
                        "    `category_name` STRING ,\n" +
                        "    `pet_type` STRING,\n" +
                        "    `is_common` bigint,\n" +
                        "    `age_stage` string\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dim_pet_categorys',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dwd-user_product_log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );


//        tabEnv.sqlQuery(
//                "select * from dim_pet_categorys"
//        ).execute().print();
    }


    //todo 映射到Kafka消息队列
    private static void createInputTable(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "CREATE TABLE `dwd_user_product_log`(\n" +
                        "    `user_id` bigint  COMMENT '用户ID',\n" +
                        "    `behavior_id` STRING ,\n" +
                        "    `behavior_type` STRING,\n" +
                        "    `category_id` bigint,\n" +
                        "    `item_id` string,\n" +
                        "    `behavior_time` timestamp,\n" +
                        "    `age_related` bigint,\n" +
                        "    `tags` STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dwd_user_pet_behavior',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dwd-user_product_log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );

//        tabEnv.sqlQuery(
//                "select * from dwd_user_product_log"
//        ).execute().print();

    }







    //todo 1-表的执行环境
    private static TableEnvironment getTableEnv() {
        //1-环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        //2-配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        //3-返回对象
        return tabEnv;
    }
}
